package com.qit.prototype.flink.performance;

import java.util.List;
import java.util.Map;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import com.qit.prototype.flink.entity.Tick;

/**
 * Listen 100000 events and response the last event (#100000) using pattern.
 * 
 * @author rollgan
 *
 */
public class TestCase4 {
	
	public TestCase4() throws Exception {
		

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		
		
		DataStreamSource<String> dataStreamSource = env
				.socketTextStream("localhost", 11002, "\0");
		
		DataStream<Tick> ds = dataStreamSource.map(new MapFunction<String, Tick>() {

			private static final long serialVersionUID = -8492269828465860764L;

			@Override
			public Tick map(String evt) throws Exception {
				String[] fields = evt.split("\\|");
				Tick tick = new Tick();
				tick.setSymbol(fields[1]);
				tick.setPrice(Double.parseDouble(fields[2]));
				tick.setQty(Double.parseDouble(fields[3]));
				tick.setTimestamp(Long.parseLong(fields[4]));
				return tick;
			}
			
		})
		.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tick>() {

			private static final long serialVersionUID = 5879473254338500372L;

			private long currentMaxTimestamp = 0l;

			@Override
			public long extractTimestamp(Tick element,
					long previousElementTimestamp) {
				long ts = element.getTimestamp();
				currentMaxTimestamp = Math.max(currentMaxTimestamp, ts);
				return ts;
			}

			@Override
			public Watermark getCurrentWatermark() {
//				if (currentMaxTimestamp > 100) {
//					return new Watermark(currentMaxTimestamp - 100);
//				}
//				else {
					return new Watermark(currentMaxTimestamp + 10);
//				}
//				return null;
			}
		});
		
		Pattern<Tick, ?> pattern = Pattern.<Tick>begin("start")
				.where(new SimpleCondition<Tick>() {

					private static final long serialVersionUID = -7850969134419767514L;
		
					@Override
					public boolean filter(Tick tick) throws Exception {
//						System.out.println(tick.getQty());
//						return true;
						if (tick.getQty() == 100000) {
							return true;
						}
						return false;
					}
			
				});
//				.followedBy("end");
//				.within(Time.milliseconds(100));
		
		PatternStream<Tick> patternStream = CEP.pattern(ds, pattern);
		
		patternStream.process(new PatternProcessFunction<Tick, Tick>() {

			private static final long serialVersionUID = 1503737342639825382L;

			@Override
			public void processMatch(
					Map<String, List<Tick>> arg0,
					org.apache.flink.cep.functions.PatternProcessFunction.Context arg1,
					Collector<Tick> arg2) throws Exception {
//				if (arg0.get("start").get(0).getQty() <= 40000)
//					return;
				System.out.println(System.currentTimeMillis());
				System.out.println(arg0.get("start").get(0).getQty());
				
			}
			
		});
		
        env.execute("Performance");
		
	}

}
